package com.ahaha.myBy


import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object aggregateByKeyTest {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setAppName(s"${combinerByKeyTest.getClass.getSimpleName}")
            .setMaster("local[*]")
        val sc = new SparkContext(conf)
        //aggregateByKeyOps(sc)
        //test1(sc)
        //aggregateByKey(sc)
        test2(sc)
        sc.stop()
    }

    def test2(sc: SparkContext): Unit = {
        val lines = sc.parallelize(List(
            "hello you",
            "hello me",
            "hello lan lan"
        ))

        val pairs = lines.flatMap(_.split("\\s+")).map((_, 1))

        def mergeValue(sum: Int, num: Int): Int = {
            sum + num
        }

        def mergeCombiners(sum1: Int, sum2: Int): Int = {
            sum1+sum2
        }

        val res = pairs.aggregateByKey(0)(mergeValue, mergeCombiners)
        println("aggregateByKey后的结果")
        res.foreach(println)
    }


    def aggregateByKey(sc: SparkContext): Unit = {
        case class Student(id: Int, name: String, province: String)

        var stuRDD = sc.parallelize(List(
            Student(1, "张三", "天津"),
            Student(4, "李四", "湖南"),
            Student(10, "王五", "天津"),
            Student(6, "赵六", "湖南"),
            Student(2, "田七", "山西"),
            Student(3, "赵八", "山东"),
            Student(7, "孙九", "山西"),
            Student(5, "郑十", "湖南"),
            Student(10087, "熊大", "山西"),
            Student(8, "熊二", "山东"),
            Student(10086, "伍六七", "湖南"),
            Student(10089, "雷电俱散", "山东")
        ), 3)

        val p2stu: RDD[(String, Student)] = stuRDD.map(stu => (stu.province, stu))


        def mergeValue(stus: Array[Student], stu: Student): Array[Student] = {
            println("同一个分区下的相同key的聚合")
            stus.+:(stu)
        }

        def mergeCombiners(stus1: Array[Student], stus2: Array[Student]): Array[Student] = {
            println("")
            stus1 ++ stus2
        }

        val ret = p2stu.aggregateByKey(Array[Student]())(mergeValue, mergeCombiners)

        ret.foreach { case (province, students) => {
            println(s"${province}对应的学生有：${students.toList}")
        }
        }
    }
}
